package com.kafka.test.broker.kafka;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import java.util.Properties;


/**
 * 1. kafka-client
 * 2. vert.x-kafka-client : https://vertx.io/docs/vertx-kafka-client/java/#_using_the_vert_x_kafka_client
 */

/**
 * attention!!!
 * example: here
 * https://docs.confluent.io/clients-kafka-java/current/overview.html
 * ProducerConfig
 * ConsumerConfig
 * KafkaProducer
 * KafkaConsumer
 *
 * <p>
 * only java config
 * user kafka-client
 * <p>
 * https://docs.spring.io/spring-kafka/docs/2.5.1.RELEASE/reference/html/#a-very-very-quick-example
 */
@Configuration
@ConfigurationProperties(prefix = "kafka-config")
@Slf4j
@Data
public class KafkaConfig {


  /**
   * kafka bootstrap-servers
   */
  private List<String> bootstrapServers;

  /**
   *
   */
  private String clientId;

  /**
   *
   */
  private List<Topic> topics;

  /**
   * 在发送数之前，创建 topic
   * <p>
   * 在生产者向topic发送之前，如果不存在，则将会自动创建topic 和partition
   * <p>
   * 这里最好进行 手动控制，直接在发送之前，程序控制 创建 topic , partition
   * <p>
   * <p>
   * https://my.oschina.net/u/2464371/blog/3107324
   * topic--N-->partition--N-->replication
   * <p>
   * https://stackoverflow.com/questions/27036923/how-to-create-a-topic-in-kafka-through-java
   * bootstrap.servers=localhost:9092
   * group.id=test
   * enable.auto.commit=true
   * auto.commit.interval.ms=1000
   * key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
   * value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
   */


  /**
   * https://docs.confluent.io/clients-kafka-java/current/overview.html
   *
   * @throws UnknownHostException
   */
  private void producer() throws UnknownHostException {
    Properties config = new Properties();
    config.put(ProducerConfig.CLIENT_ID_CONFIG, InetAddress.getLocalHost().getHostName());
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "host1:9092,host2:9092");
    config.put(ProducerConfig.ACKS_CONFIG, "all");//leader and followers ack
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, new StringSerializer());


    config.put(ProducerConfig.RETRIES_CONFIG, 0);
    config.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    config.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);


//        config.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        config.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        config.put("group.id", "my_group");
//        config.put("auto.offset.reset", "earliest");
//        config.put("enable.auto.commit", "false");


    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");//
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "test_group"); //

  }


}
